// ====================== HEADER FILES ======================
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <stdbool.h>
#include <sys/mman.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/stat.h> /* For mode constants */
#include <fcntl.h> /* For O_* constants */
#include <errno.h>
#include <math.h>
#include <time.h>

// ================= DATA TYPE DECLARATIONS =================
#define SECTOR_SIZE 512
typedef unsigned char byte;

typedef struct {
    int sectorNum;
    bool served;
    double processTime;
} Request;

typedef struct {
	int sector_num;
	int served;  //0 or 1
	int finished; //0 or 1
	byte data[SECTOR_SIZE];  //Each character is byte sized
} RingBuffer;

typedef struct {
	pthread_t thread;
	Request *requests;
	int size;
} Thread;


// =================== GLOBAL VARIABLES ===================
FILE *fp;
FILE *c_fp;

FILE *sectorNumbers;
FILE *returnValues;

int shm_id;
char filename[255];
RingBuffer *buffer;

int numThreads = 0;
int numReq = 0;
Thread *threads;
Request *requests;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
int served_count = 0;

// ================== FUNCTION PROTOTYPES ==================
void createSharedMemory();
void generateRequests(int);
void divideRequests();
int runThreads();
void printData();
void *threadFunc(void*);
void sendRequest(Request*);


int main(int argc, char *argv[])
{
    int i;
    int friend_pid;
    int num_sectors = 0;
    int count = 0;
    char sectorFile[50], readFile[50], pid[50];
    sprintf(pid, "%d", getpid());
    
	if (argc != 3) {
	        printf("Invalid number of arguments\n");
	} else {
		// initialize variables
		numThreads = atoi(argv[1]);
	    numReq = atoi(argv[2]);
	    threads = (Thread *)malloc(numThreads*sizeof(*threads));
	    requests = (Request *)malloc(numReq*sizeof(*requests));

	    for(i = 0; i < numThreads; i++) {
	    	threads[i].size = 0;
	    	threads[i].requests = (Request *)malloc(((numReq/numThreads) + 1)*sizeof(*requests));
	    }

	    // create file for sector numbers
	    strcpy(sectorFile, "sectors.");
	    strcat(sectorFile, pid);
	    sectorNumbers = fopen(sectorFile, "a+");
	    chmod(sectorFile, 00755);

	    // create file for read results
	    strcpy(readFile, "read.");
	    strcat(readFile, pid);
	    returnValues = fopen(readFile, "a+");
	    chmod(readFile, 00755);

	    createSharedMemory();

	   	// Send signal to server for registration an inform location of ring buffer
	    fp = fopen("ch_id", "r");
	    c_fp = fopen("clients", "w");

	   	fprintf(c_fp, "Client %d has arrived with %s\n", (int)getpid(), filename);
	   	if(fscanf(fp, "%d, %d", &friend_pid, &num_sectors) != 2) {
	   	   	printf("Reading ch_id file failed\n");
	   	}

	   	kill(friend_pid, SIGUSR1); // Sends this signal to all processes it has perms to, hopefully the server

//	   	printf("Signal sent\n");
    	fclose(c_fp);

	    generateRequests(num_sectors);
	    divideRequests();
	    runThreads();
	    printData();

	    close(shm_id);
	    fclose(sectorNumbers);
	    fclose(returnValues);
	}

	return 0;
}

/**
 * Creates and maps shared memory or ring buffer for client
 */
void createSharedMemory() {
	// Map memory
	sprintf(filename, "/Ring_buff_%d", (int)getpid());	// filename will be the name of shared memory object

	// Create shared memory or ring buffer for client
	if((shm_id = shm_open(filename, O_RDWR|O_CREAT, 0777)) < 0) {
		perror("shm_open");
		printf("client shm_open failed\n");
		exit(1);
	}

	// Allocates memory for shared memory object
	if(ftruncate(shm_id, sizeof(RingBuffer)) == -1)
		printf("truncate failed\n");

	// Maps the shared memory object to the process's address space
   	buffer = mmap(0,  				// Let kernel find memory
   			sizeof(RingBuffer),  	// Length
		   	PROT_WRITE|PROT_READ, 	// Protection
		   	MAP_SHARED ,			// Flags
	        shm_id,					// fid
	        0);						// Offset

   	if(buffer == MAP_FAILED) {
	   	fprintf( stderr, "client map failed: %s\n",
	               strerror( errno ) );
	    exit(1);
	}

   	//buffer->sector_num = (getpid()%10);
   	buffer->served = 1;
   	buffer->finished = 0;

   	// Flush cached data in address space to the shared memory object, invalidating everyone else's cached data
   	msync(buffer, sizeof(RingBuffer), MS_SYNC|MS_INVALIDATE);
} // end createSharedMemory


/**
 * This function sends the number of request to the server
 * and the server returns the first & last sector number of
 * the virtual disk image.
 */
void generateRequests(int numSectors) {
    int i;
    int firstSectorNum = 0;
    int lastSectorNum = (numSectors-1);

    // randomly generate sector numbers within the range of sectors on
    // the virtual disk image for each request
    srand(time(NULL));
    for (i = 0; i < numReq; i++) {
        requests[i].sectorNum = (rand() % ((lastSectorNum - firstSectorNum) +1 ) + firstSectorNum);
        requests[i].served = false;
//        printf("\nrequests[%d]: %d", i, requests[i].sectorNum);
    }
} // end generateRequests


/**
 * divide the # of requests evenly amongst the threads
 * if there is a remainder of requests, add one to as many
 * threads it covers
 */
void divideRequests() {
	int j, k;
    int reqPerThread = numReq / numThreads;

    k = -1;
    int last_idx = reqPerThread * numThreads;

    // add reqPerThread to each thread
    for(j = 0; j < last_idx; j++) {
    	if(j % reqPerThread == 0)
    		k++;

	    threads[k].requests[j % reqPerThread] = requests[j];
	    threads[k].size++;
   	} // end for j

    // if there are a remainder of requests separate them amongst
    // the threads
    if(last_idx != numReq) {
    	k = 0;
    	for(j = last_idx; j < numReq; j++) {
    		threads[k].requests[reqPerThread] = requests[j];
    		threads[k].size++;
    		k++;
    	} // end for j
    } // end if there is a remainder
} // end divideRequests


/**
 * Run each thread in the client
 */
int runThreads() {
	int j, rc;
	unsigned long k;
	pthread_mutex_init(&mutex, NULL);

	for(k = 0; k < numThreads; k++) {
		if((rc = pthread_create(&threads[k].thread, NULL, threadFunc, (void *)k))) {
			fprintf(stderr, "error: pthread_create, rc: %d\n", rc);
			return EXIT_FAILURE;
		} // end if
    } // end for k

	for (j = 0; j < numThreads; j++) {
		for(k = 0; k < threads[j].size; k++) {
			pthread_join(threads[j].thread, NULL);
	    }
	}

	buffer->finished = 1;
	msync(buffer, sizeof(RingBuffer), MS_SYNC|MS_INVALIDATE);
//	printf("\ncount = %d", served_count);
	return EXIT_SUCCESS;
} // end runThreads


/**
 * This function is executed every time
 */
void *threadFunc(void* param) {
	unsigned long threadNum = (unsigned long) param;
	int i;

	for(i = 0; i < threads[threadNum].size; i++) {
		sendRequest(&threads[threadNum].requests[i]);
	}
}


/**
 * If the mutex is unlocked, a request is sent to the
 * server when the ring buffer is free.
 */
void sendRequest(Request* r) {
	byte *value;
	struct timeval current_time;
	// store request sector number in file
	fprintf(sectorNumbers, "%d\n", r->sectorNum);

	pthread_mutex_lock(&mutex);
	//clock_t startTime = clock();

		gettimeofday(&current_time,NULL);
		long long before = 1000*((current_time.tv_sec * 1000000) + (current_time.tv_usec));
	buffer->sector_num = r->sectorNum;
	buffer->served = 0;
	buffer->finished = 0;
	msync(buffer, sizeof(RingBuffer), MS_SYNC|MS_INVALIDATE);
	//printf("\nServer updated to %d from %ld", buffer->sector_num, getpid());

	while(buffer->served == 0) {} // Wait until server updates shared memory


	value = buffer->data;
	served_count++;

	pthread_mutex_unlock(&mutex);
	//clock_t endTime = clock();
	gettimeofday(&current_time,NULL);
	long long after = 1000*((current_time.tv_sec * 1000000) + (current_time.tv_usec));

	//r->processTime = ((double)(endTime - startTime) / CLOCKS_PER_SEC) * 1000000000; // convert to nanoseconds
	r->processTime = after - before;
	//printf("\n(sendRequest) time = %lf", r->processTime);

	// store request return value in file
	fprintf(returnValues, "%s\n", value);
} // end sendRequest

/**
 * Print maximum, average, & minimum time taken to process
 * each request, the standard deviation, and the processed read
 * requests per second on the screen.
 */
void printData() {
	int j, k;
	long int avgTime, total;
	double stanDev, reqPerSec;
    long int maxTime = 0, minTime = 0;

    // get max, min, and average process time
    total = 0;
    for(j = 0; j < numThreads; j++) {
    	Thread t = threads[j];
    	for(k = 0; k < t.size; k++) {
    		total += t.requests[k].processTime;

    		if(maxTime < t.requests[k].processTime)
    			maxTime = t.requests[k].processTime;
    		if(minTime > t.requests[k].processTime || minTime == 0)
    			minTime = t.requests[k].processTime;
    	} // end for k
    } // end for j
    avgTime = total / numReq;

    // calculate request per second
    reqPerSec = (numReq / (double)total) * 1000000000;

    // calculate standard deviation
    total = 0;
    for(j = 0; j < numThreads; j++) {
        Thread t = threads[j];
       	for(k = 0; k < t.size; k++) {
       		total += pow((avgTime - t.requests[k].processTime), 2);
       	} // end for k
    } // end for j
    stanDev = sqrt(total / numReq);

    printf("\nMax | ");
    printf("Avg | ");
    printf("Min | ");
    printf("Stan. Dev. | ");
    printf("Req/Sec");

    printf("\n%ld|", maxTime);
    printf("%ld|", avgTime);
    printf("%ld|", minTime);
    printf("%3.1lf|", stanDev);
    printf("%3.1lf\n", reqPerSec);

} // end printData
